package com.match.plugin.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import com.match.input.OrderData;
import cn.hutool.json.JSONUtil;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
/**
 * kafka工具类
 */
public class Kafka {
	
	//ticp,map-key,group.id 共用名称
    public static final String matchTrade = "matchTrade";
    public static final String orderChange = "orderChange";
    public static final String tradeDepth = "tradeDepth";
    public static final String revokeResult = "revokeResult";
    
    public static final String matchOrder = "matchOrder";
    public static final String revokeOrder = "revokeOrder";
    
	public static final ConcurrentHashMap<String, KafkaProducer<Long, String>> producerMap = new ConcurrentHashMap<String, KafkaProducer<Long, String>>();//生产者
	public static final ConcurrentHashMap<String, KafkaConsumer<Long, String>> consumerMap = new ConcurrentHashMap<String, KafkaConsumer<Long, String>>();//消费者
	
	
	//消费者消费开关
	
	
	/**
     * 添加生产者
     * @param name
     * @param servers
     * @param keySerializer
     * @param valueSerializer
     */
    public static void addProducer(String producerName,Properties props) {
        if (!producerMap.containsKey(producerName)) {
            KafkaProducer<Long, String> producer = new KafkaProducer<Long, String>(props);
            producerMap.put(producerName, producer);
        }
    }
    
    /**
	 * 添加消费者
	 */
    public static KafkaConsumer<Long, String> addConsumer(String consumerName,String topic,Properties props) {
    	props.put("group.id", consumerName);
    	
    	if (consumerMap.containsKey(consumerName)) {
    		return consumerMap.get(consumerName);
    	}else {
    		KafkaConsumer<Long, String> consumer = new KafkaConsumer<Long, String>(props);
	    	consumer.subscribe(Collections.singletonList(topic));//订阅TOPIC
	    	consumerMap.put(consumerName, consumer);
	    	return consumerMap.get(consumerName);
    	}
    }
	
    /**
     * 获取一个消费者
     */
    public static KafkaConsumer<Long, String> getConsumer(String name) {
		return consumerMap.get(name);
	}
	
	/**
	 * 获取一个生产者
	 * @param name
	 * @return
	 */
	public static KafkaProducer<Long, String> getProducer(String name) {
		return producerMap.get(name);
	}
	
	
	/**
	 * 生产者发送消息
	 * @param <T>
	 * @param <T>
	 */
	public static <T> Future<RecordMetadata> send(String producerName,String topicName,Long key,T Obj) {
		String jsonStr = JSONUtil.toJsonStr(Obj);
		return producerMap.get(producerName).send(new ProducerRecord<Long, String>(topicName, key, jsonStr));
	}
	
	
	/**
	 * 消费者消费(撮合)
	 * @param consumer1
	 */
	public static void matchOrderConsumer(KafkaConsumer<Long, String> consumer1){
    	try {
		    while (true) {
		        // 100 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
		    	ConsumerRecords<Long, String> records = consumer1.poll(Duration.ofMillis(100)); 
		        // poll 返回一个记录列表。
		        // 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。
		        for (ConsumerRecord<Long, String> record : records) {
//			        System.out.println("customer="+record.key()+", country="+record.value());			        
			        OrderData.new_order(record.value());
			    }
		        /*consumer.commitSync();//同步*/
		    } 
		} finally {
		    // 关闭消费者,网络连接和 socket 也会随之关闭，并立即触发一次再均衡
			consumer1.close();
		}
    }
	
	/**
	 * 消费者消费(撤销)
	 * @param consumer2
	 */
	public static void revokeOrderConsumer(KafkaConsumer<Long, String> consumer2){
    	try {
		    while (true) {
		        // 100 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
		    	ConsumerRecords<Long, String> records = consumer2.poll(Duration.ofMillis(Long.MAX_VALUE)); 
		        // poll 返回一个记录列表。
		        // 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。
		        for (ConsumerRecord<Long, String> record : records) {
			        //System.out.println("topic="+record.topic()+", partition="+record.partition()+", offset="+record.offset()+", customer="+record.key()+", country="+record.value());			        
			        OrderData.cancel_order(record.value());
			    }
		        /*consumer.commitSync();//同步*/
		    } 
		} finally {
		    // 关闭消费者,网络连接和 socket 也会随之关闭，并立即触发一次再均衡
			consumer2.close();
		}
    }
	
	
}
